package com.xbai.spark.recall.engine.training

import breeze.linalg.split
import breeze.numerics.{pow, sqrt}
import org.apache.spark.sql.functions.{desc, udf}
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.UserDefinedFunction

import scala.collection.mutable

/**
  * user-item-rating
  * 计算物品之间相同用户评分乘机的和做分子
  * 物品评分平方和的开方的乘积做分母
  * (x1*y1+x2*y2......)/|X|*|Y|
  * |X| = sqrt(x1^2 + x2^2 + x3^2)
  * |Y| = sqrt(y1^2 + y2^2 + y3^2)
  *
  * @author xbai
  * @Date 2021/1/22
  */
class ItemCF {

  /**
    * 计算物品之间相同用户评分乘机的和做分子
    * (x1*y1+x2*y2......)
    *
    * @param userItemRating userId-itemId-rating
    * @param spark          spark环境
    * @return itemId-itemId1-ratingSumPro
    */
  def getItemSameUserProductSum(userItemRating: DataFrame, spark: SparkSession): DataFrame = {
    val userItemRatingCopy: DataFrame = userItemRating.selectExpr("userId as userId1", "itemId as itemId1", "rating as rating1")
    val itemRating: DataFrame = userItemRating.join(userItemRatingCopy, userItemRatingCopy("userId1") === userItemRating("userId"))
      .filter("cast(itemId as long) != cast(itemId1 as long)")
      .selectExpr("itemId", "itemId1", "rating*rating1 as ratingProduct")
    val itemRatingSum: DataFrame = itemRating.groupBy("itemId", "itemId1")
      .sum("ratingProduct")
      .withColumnRenamed("sum(ratingProduct)", "ratingSumPro")
    itemRatingSum.show(3)
    itemRatingSum
  }

  /**
    * 物品评分平方和的开方
    * |X| = sqrt(x1^2 + x2^2 + x3&#94;2)
    *
    * @param userItemRating userId-itemId-rating
    * @param spark          spark环境
    * @return itemId-sqrtRatingSum: itemId-rating 评分的模
    */
  def getItemSqrtRatingSum(userItemRating: DataFrame, spark: SparkSession): DataFrame = {
    import spark.implicits._
    val itemRating: DataFrame = userItemRating.rdd.map(x => (x(1).toString, x(2).toString))
      .groupByKey()
      .mapValues(x => sqrt(x.toArray.map(rating => pow(rating.toDouble, 2)).sum))
      .toDF("itemId", "sqrtRatingSum")
    itemRating.show(3)
    itemRating
  }

  /**
    *
    * @param itemRating itemId-sqrtRatingSum
    * @param itemSameUserRating  itemId-itemId1-ratingSumPro
    * @param spark spark环境
    * @return itemId-itemId1-itemSimilar
    */
  def itemSimilarity(itemRating: DataFrame, itemSameUserRating: DataFrame, spark: SparkSession): DataFrame = {
    val itemRatingCopy: DataFrame = itemRating.selectExpr("itemId as itemId1", "sqrtRatingSum as sqrtRatingSum1")
    val itemSimilar: DataFrame = itemSameUserRating.join(itemRating, "itemId")
      .join(itemRatingCopy, "itemId1")
      .selectExpr("itemId", "itemId1", "ratingSumPro/(sqrtRatingSum*sqrtRatingSum1) as itemsim")
    itemSimilar.show(3)
    itemSimilar.sort(desc("itemId"), desc("itemsim"))
      .filter("itemsim > 0")
  }

  def itemJaccardSimilarity(userItemRating: DataFrame, spark: SparkSession): DataFrame = {
    import org.apache.spark.sql.functions._
    val itemUsers: DataFrame = userItemRating.groupBy("itemId")
      .agg(collect_set("userId"))
      .withColumn("users", concat_ws(",", col("collect_set(userId)")))
      .withColumn("users", functions.split(col("users"), ","))
      .withColumn("id", monotonically_increasing_id)
    itemUsers.show(3)
    
    val itemUsersCopy: DataFrame = itemUsers.selectExpr("itemId as itemId1", "users as users1", "id as id1")
    val itemSimilar: Dataset[Row] = itemUsers.join(itemUsersCopy, itemUsers("id") =!= itemUsersCopy("id1"))
      .drop("id")
      .withColumn("itemsim", jaccard(col("users"), col("users1")))
      .selectExpr("itemId", "itemId1", "itemsim")
      .sort(desc("itemId"), desc("itemsim"))
      .filter("itemsim > 0")
    itemSimilar.write.mode("overwrite").saveAsTable("itemJaccardSimilar")
    itemSimilar
  }

  /**
    * jaccard 距离
    * A & B / A | B = A & B / (A + B - (A & B))
    */
  val jaccard: UserDefinedFunction = udf((users: mutable.WrappedArray[String], usersCopy: mutable.WrappedArray[String]) => {
    val usersSet: Set[String] = users.toSet
    val usersCopySet: Set[String] = usersCopy.toSet
    val set: Set[String] = usersSet & usersCopySet
    var value: Float = 0.0f
    if (set.nonEmpty) {
      value = set.size / (usersSet.size + usersCopySet.size - set.size)
    }
    value
  })
}
